package com.at.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 * @author cdhuangchao3
 * @date 2023/5/19 10:02 PM
 */
object Spark05_Req1_HotCategoryTop10SessionAnalysis {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    // Q1: rdd重复使用
    // Q2: 性能可能较低
    // Q3: 存在大量reduceByKey， 聚合算子，Spark会提供优化

    // 1、读取原始数据
    val actionRDD = sc.textFile("datas/user_visit_action.txt")
    actionRDD.cache()
    val top10Ids:Array[String] = top10Category(actionRDD)

    // 过滤原始数据
    val filterActionRDD = actionRDD.filter(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          top10Ids.contains(datas(6))
        } else {
          false
        }
      }
    )
    // 2、根据品类ID、sessionId进行点击量统计
    val reduceRDD = filterActionRDD.map {
      action => {
        val datas = action.split("_")
        ((datas(6), datas(2)), 1)
      }
    }.reduceByKey(_ + _)

    // 3、将统计结果进行结构转换
    val mapRDD = reduceRDD.map {
      case ((cid, sid), num) => {
        (cid, (sid, num))
      }
    }

    // 4、相同的品类进行分组
    val groupRDD = mapRDD.groupByKey()

    // 5、将分组后的数据进行点击量排序，取前10名
    val resultRDD = groupRDD.mapValues(
      iter => {
        val resultRDD = iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
        resultRDD
      }
    )

    resultRDD.collect().foreach(println)

    sc.stop()
  }

  def top10Category(actionRDD: RDD[String]): Array[String] = {
    val flatRDD = actionRDD.flatMap(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.map(cid => (cid, (0, 1, 0)))
        } else if (datas(10) != "null") {
          val cids = datas(10).split(",")
          cids.map(cid => (cid, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )

    // 3、聚合
    val analysisRDD = flatRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    // 4、降序排列取前10
    analysisRDD.sortBy(_._2, false).take(10).map(_._1)
  }
}
